package com.atguigu.edu.realtime.util;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.common.EduConfig;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.List;

public class HbaseUtil {
    public static Connection getHbaseConnect(){
        System.out.println("获取Hbase连接");
        Configuration conf = new Configuration();
        conf.set("hbase.zookeeper.quorum", EduConfig.Zookeeper);
        try {
            return ConnectionFactory.createConnection(conf);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }


    public static void closeHbaseConnect(Connection conn){
        if (!conn.isClosed() && conn != null){
            System.out.println("关闭Hbase连接");
            try {
                conn.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

   //根据rowkey获取hbase表中的一行数据
    public static JSONObject getDimInfoFromHbase(Connection conn,String nameSpace,String tableName,String rowkey){
        TableName tableNameObj = TableName.valueOf(nameSpace, tableName);
        try (Table table = conn.getTable(tableNameObj)){
            Get get = new Get(Bytes.toBytes(rowkey));
            Result result = table.get(get);
            List<Cell> cells = result.listCells();
            if(cells != null && cells.size() > 0){
                //创建一个新的json对象，用于封装查询结果
                JSONObject jsonObj = new JSONObject();
                for (Cell cell : cells) {
                    String columnName = Bytes.toString(CellUtil.cloneQualifier(cell));
                    String columnValue = Bytes.toString(CellUtil.cloneValue(cell));
                    jsonObj.put(columnName,columnValue);
                }
                return jsonObj;
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }


    //从Hbase创建表
    public static void createHbaseTables(Connection conn,String nameSpace,String tableName, String... families){
        //判断是否有列族
        if (families == null || families.length < 1){
            throw new RuntimeException("在建表列族必须提供");
        }
        try(Admin admin = conn.getAdmin()){
            //判断要创建的表是否存在
            TableName tableNameObj = TableName.valueOf(nameSpace, tableName);
            if (admin.tableExists(tableNameObj)){
                System.out.println("要创建的"+ nameSpace + ":" + tableName + "已存在");
                return;
            }
            System.out.println("在hbase创建" + nameSpace + ":" + tableName + "表");
            TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(tableNameObj);
            for (String family : families) {
                ColumnFamilyDescriptor columnFamilyDescriptor = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(family)).build();
                tableDescriptorBuilder.setColumnFamily(columnFamilyDescriptor);
            }
            admin.createTable(tableDescriptorBuilder.build());
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    //从Hbase删除表
    public static void deleteHbaseTables(Connection conn,String nameSpace,String tableName){
        try (Admin admin = conn.getAdmin()) {
            //判断要删除的表是否存在
            TableName tableNameObj = TableName.valueOf(nameSpace, tableName);
            if (!admin.tableExists(tableNameObj)) {
                System.out.println("要删除的" + nameSpace + ":" + tableName + "不存在");
                return;
            }
            System.out.println("从hbase中删除" + nameSpace + ":" + tableName + "表");
            admin.disableTable(tableNameObj);
            admin.deleteTable(tableNameObj);
        } catch (Exception e) {
            e.printStackTrace();
        }

    }


    //根据rowkey从Hbase中删除一行
    public static void delRow(Connection conn, String nameSpace, String tableName, String rowKey) throws IOException{
        TableName tableNameObj = TableName.valueOf(nameSpace, tableName);
        Table table = conn.getTable(tableNameObj);
        Delete delete = new Delete(Bytes.toBytes(rowKey));
        table.delete(delete);
        table.close();
    }

    //根据rowkey从Hbase中添加一行
    public static void putRow(Connection conn, String nameSpace, String tableName, String rowKey, String family, String[] columns, String[] values) throws IOException {
        TableName tableNameObj = TableName.valueOf(nameSpace, tableName);
        Table table = conn.getTable(tableNameObj);
        Put put = new Put(Bytes.toBytes(rowKey));
        for (int i = 0; i < columns.length; i++) {
            String columnName = columns[i];
            String columnValue = values[i];
            if (columnValue != null) {
                put.addColumn(Bytes.toBytes(family), Bytes.toBytes(columnName), Bytes.toBytes(columnValue));
            }
        }
        table.put(put);
        table.close();
    }
    //获取支持异步操作的连接对象
    public static AsyncConnection getAsyncConnection(){
        Configuration conf = new Configuration();
        // 配置 Zookeeper
        conf.set("hbase.zookeeper.quorum", "hadoop102,hadoop103,hadoop104");
        try {
            return ConnectionFactory.createAsyncConnection(conf).get();
        } catch (Exception e) {
            throw new RuntimeException();
        }
    }

    //关闭支持异步操作的连接对象
    public static void closeAsyncConnection(AsyncConnection conn){
        if(conn != null && !conn.isClosed()){
            try {
                conn.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 以异步的方式从hbase维度表中查询维度数据
     * @param conn          异步连接客户端
     * @param nameSpace     表空间名
     * @param tableName     表名
     * @param rowkey        主键
     * @return      将查询出来的维度数据封装为一个json对象
     */
    public static JSONObject getDimInfoFromHbaseByAsync(AsyncConnection conn, String nameSpace, String tableName, String rowkey){
        try {
            TableName tableNameObj = TableName.valueOf(nameSpace, tableName);
            AsyncTable<AdvancedScanResultConsumer> asyncTable = conn.getTable(tableNameObj);
            Get get = new Get(Bytes.toBytes(rowkey));
            Result result = asyncTable.get(get).get();
            List<Cell> cells = result.listCells();
            if(cells != null && cells.size() > 0){
                JSONObject jsonObj = new JSONObject();
                for (Cell cell : cells) {
                    String columnName = Bytes.toString(CellUtil.cloneQualifier(cell));
                    String columnValue = Bytes.toString(CellUtil.cloneValue(cell));
                    jsonObj.put(columnName,columnValue);
                }
                return jsonObj;
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }

    public static void main(String[] args) {
        Connection conn = getHbaseConnect();
        JSONObject dimJsonObj = getDimInfoFromHbase(conn, EduConfig.HBASE_NAMESPACE, "dim_test_paper", "1130");
        System.out.println(dimJsonObj);
        closeHbaseConnect(conn);
    }
}
